package com.ls.rabbitmqService.direct;

import com.ls.rabbitmqService.utils.ConnectionFactoryUtil;
import com.rabbitmq.client.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;

public class ReceivedMessageDirect {


    private static final Logger logger = LoggerFactory.getLogger(ReceivedMessageDirect.class);

    public static void receiveMessage(boolean durable,String exchanges_name,String queue_name,String routingKey){
        Connection connection =null;
        Channel channel=null;
        try {
            connection=ConnectionFactoryUtil.getRabbitConnection();
            channel = connection.createChannel();
            channel.exchangeDeclare(exchanges_name, BuiltinExchangeType.DIRECT,durable);
            channel.queueBind(queue_name,exchanges_name,routingKey);
            logger.info(String.format("ReceivedMessageDirect exchange:%s,queueName:%s,severity:%s",exchanges_name,queue_name,routingKey));
            logger.info(" ReceivedMessageDirect wait.....");

            final Channel channelCopy=channel;
            Consumer consumer=new DefaultConsumer(channelCopy){
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                    //采用手动确认
                    channelCopy.basicAck(envelope.getDeliveryTag(),false);
                    String message=new String(body,"utf-8");
                    if(logger.isDebugEnabled())
                        logger.debug(String.format("ReceivedMessageDirect [x4] routingKey:%s message:%s",envelope.getRoutingKey(),message));
                }
            };

            //这样会持续的接受消息  false 代表需要手动确认消息
            channel.basicConsume(queue_name,false,consumer);

        } catch (IOException e) {
           logger.error(e.getMessage(),e);
        } finally{

        }

    }

}

